package saer.send.now.channel;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;

import com.google.appengine.repackaged.org.json.JSONException;
import com.google.appengine.repackaged.org.json.JSONObject;

public class Channel {
    private static final Logger  log    = Logger.getLogger(Channel.class.getName());
    private static AtomicInteger nextId = new AtomicInteger(1337);

    private OutputStream         output;
    private String               channelId;
    private State                state;

    public Channel(String id, State state) {
        this.channelId = id;
        this.state = state;
    }

    public synchronized void upload(InputStream input, int lenght, String separator) {
        if (this.output == null) {
            throw new RuntimeException("Output for upload has not been set on channel, id=" + this.channelId);
        }

        ChannelInputStream channelInput = new ChannelInputStream(input);

        try {
            int headerLength = channelInput.skippHeaders();
            this.state.setTotalBytesToSend(lenght - headerLength - separator.length() - 8);

            byte[] buffer = new byte[1024];
            int length;
            while (true) {
                length = (int) Math.min(buffer.length, this.state.getBytesLeft());
                if (length == 0) {
                    break;
                }
                length = channelInput.read(buffer, length);
                if (length == -1) {
                    break;
                }
                this.state.incBytesSent(length);
                this.output.write(buffer, 0, length);
            }
        } catch (IOException e) {
            this.state.setError("Transaction failed, " + e.getMessage());
            throw new RuntimeException(e);
        } finally {
            try {
                this.output.close();
                channelInput.close();
            } catch (IOException e) {
                // Do not care we are done
                log.log(Level.WARNING, "Failed to close streams, {0}", e.getMessage());
            }

            this.signal();
        }
    }

    public synchronized void download(OutputStream output) {
        this.output = output;
        this.state.startUpload();

        this.sleep();
    }

    public State getState() {
        return this.state;
    }

    private synchronized void signal() {
        this.notify();
    }

    private synchronized void sleep() {
        try {
            this.wait();
        } catch (InterruptedException e) {
            log.log(Level.INFO, "channel sleep has been interrupted, id:{0}, msg:{1}", new Object[] { this.channelId, e.getMessage() });
        }
    }

    public static String getNextId() {
        try {
            JSONObject idObject = new JSONObject();
            idObject.put("id", nextId.getAndIncrement());
            return idObject.toString();
        } catch (JSONException e) {
            assert (false);
            throw new RuntimeException(e);
        }
    }
}